import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.DoubleCounter;
import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo06_accumulator {


    /**
     * 默认提供的累加器 只提供数值类型
     * @param sensorDataStream
     */
    public static void accumulator_default(DataStream<Demo02_sensorReading> sensorDataStream){

        sensorDataStream
                .keyBy(Demo02_sensorReading::getSensorId)
                .map(
                        new RichMapFunction<Demo02_sensorReading, Object>() {

                            //声明一个温度的累加器和一个数量的累加器
                            DoubleCounter tempCounter = new DoubleCounter();
                            IntCounter numCounter= new IntCounter();

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                getRuntimeContext().addAccumulator("temp",tempCounter);
                                getRuntimeContext().addAccumulator("num",numCounter);
                            }

                            @Override
                            public Object map(Demo02_sensorReading value) throws Exception {

                                tempCounter.add(value.getTemperature());
                                numCounter.add(1);
                                return null;
                            }
                        }
                );
    }

    /**
     * 自定义的累加器 提供任意类型
     * @param sensorDataStream
     */
    public static void accumulator_self(DataStream<Demo02_sensorReading> sensorDataStream) {

        sensorDataStream
                .keyBy(Demo02_sensorReading::getSensorId)
                .map(
                        new RichMapFunction<Demo02_sensorReading, Object>() {

                            Demo06_selfAccumulator selfAccumulator = new Demo06_selfAccumulator();

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                getRuntimeContext().addAccumulator("self",selfAccumulator);
                            }

                            @Override
                            public Object map(Demo02_sensorReading value) throws Exception {
                                selfAccumulator.add(value.getTemperature());
                                return null;
                            }
                        }
                );
    }

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //接收String类型的原始数据
        DataStreamSource<String> dataStream = Demo02_source.source_socket(env,9999);
        //把String类型转成SensorReading类型
        DataStream<Demo02_sensorReading> sensorDataStream = Demo02_transformation.transformation_simple(dataStream);

//        accumulator_default(sensorDataStream);
        accumulator_self(sensorDataStream);

        try {
            JobExecutionResult result = env.execute("state");
            String tempAll = result.getAccumulatorResult("self");

            System.out.println("tempAll:"+tempAll);

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
